package com.xyy.saas.payment.mq;

import com.xyy.saas.payment.common.constants.MqConstant;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

/**
 * @author wangtianqi
 * @date 2019/10/24 14:55
 */
@Component
public class MqProducer implements InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(MqProducer.class);

    /**
     * 初始化rocketMQ模板
     */
    @Resource
    private RocketMQTemplate rocketMqTemplate;

    @Value("${rocketmq.producer.group}")
    private String producerGroup;
    @Value("${rocketmq.name-server}")
    private String nameServer;

    @Override
    public void afterPropertiesSet() {
        rocketMqTemplate.getProducer().setNamesrvAddr(nameServer);
        rocketMqTemplate.getProducer().setProducerGroup(producerGroup);
        rocketMqTemplate.setMessageQueueSelector(new IDHashMessageQueueSelector());
        rocketMqTemplate.getProducer().setRetryTimesWhenSendFailed(5);
        rocketMqTemplate.getProducer().setSendMsgTimeout(MqConstant.SEND_TIME_OUT);
    }

    /**
     * 设置producer负载
     */
    class IDHashMessageQueueSelector implements MessageQueueSelector {

        @Override
        public MessageQueue select(List<MessageQueue> mqs, Message msg,
                                   Object arg) {
            int id = Integer.parseInt(arg.toString());
            int size = mqs.size();
            int index = id%size;
            return mqs.get(index);
        }
    }

    /**
     * 同步消息
     * @param msg
     * @return
     */
    public SendResult syncSendMsg(String topic, String msg) {
        return rocketMqTemplate.syncSend(topic, MessageBuilder.withPayload(msg).build());
    }

    /**
     * 同步延迟消息
     * @param msg
     * @param delayLevel
     */
    public void syncSendDelayMsg(String topic, String msg, int delayLevel) {
        rocketMqTemplate.syncSend(topic, MessageBuilder.withPayload(msg).build(),MqConstant.SEND_TIME_OUT,delayLevel);
        LOGGER.info("MqProducer#syncSendDelayMsg topic:{} msg:{} delayLevel:{}", topic, msg, delayLevel);
    }


    /**
     * 异步消息
     * @param msg
     */
    public void asyncSendMsg(String topic, String msg) {

        rocketMqTemplate.asyncSend(topic, MessageBuilder.withPayload(msg).build(), new SendCallback() {
            //消息发送成功回调
            @Override
            public void onSuccess(SendResult sendResult) {
                LOGGER.info("MqProducer#onSuccess asyncSendDelayMsg status:{} sendResult:{}", sendResult.getSendStatus(), sendResult.toString());
            }

            //消息异常回调
            @Override
            public void onException(Throwable e) {
                // 同步发送一遍
                syncSendMsg(topic, msg);
                LOGGER.error("MqProducer#onSuccess asyncSendDelayMsg topic:{} msg:{}", topic, msg, e);
            }
        });
    }
}

